package cn.itcast.windows;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @author KTL
 * @version V1.0
 * @Package cn.itcast.windows
 * @date 2021/2/27 0027 14:25
 * @Copyright © 2015-04-29  One for each, and two for each
 *          演示基于Session会话窗口
 */
public class WindowSessionDemo_1_2 {
    public static void main(String[] args) throws Exception {
        //TODO 1.env
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //TODO 2.source
        final DataStreamSource<String> lines = env.socketTextStream("192.168.52.120", 9999);
        //TODO 3.transfroms
        final SingleOutputStreamOperator<CartInfo> carDS = lines.map(new MapFunction<String, CartInfo>() {
            @Override
            public CartInfo map(String value) throws Exception {
                final String[] arr = value.split(",");
                return new CartInfo(arr[0], Integer.parseInt(arr[1]));
            }
        });
        final KeyedStream<CartInfo, String> keyedDS = carDS.keyBy(CartInfo::getSensorId);
        final SingleOutputStreamOperator<CartInfo> result = keyedDS.window(ProcessingTimeSessionWindows.withGap(Time.seconds(10))).sum("count");
        //TODO 4.sink
        result.print();
        //TODO 5.execute
        env.execute();
    }
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class CartInfo{
        private String sensorId; //信号灯id
        private Integer count;//通过该信号灯车的数量
    }
}
